package com.wang.weibo.dao;

import com.wang.weibo.constants.Constants;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @Author: wang
 * @Date: 2020/1/28 17:01
 * @Description:
 */
public class HBaseDao {

    /**
     * 发布微博
     * @param uid
     * @param content
     * @throws IOException
     */
    public static void publishWeiBo(String uid, String content) throws IOException {
        Connection connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
        // 操作微博内容表
        // 获取微博内容表对象
        Table contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
        Long ts = System.currentTimeMillis();
        String rowKey = uid + "_" + ts;
        Put contPut = new Put(Bytes.toBytes(rowKey));
        contPut.addColumn(Bytes.toBytes(Constants.CONTENT_TABLE_CF), Bytes.toBytes("content"),
                Bytes.toBytes(content));
        // 插入数据
        contTable.put(contPut);
        // 操作微博收件箱表
        // 获取用户关系表对象
        Table relaTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
        // 获取当前发布微博人的fans列族数据
        Get get = new Get(Bytes.toBytes(uid));
        get.addFamily(Bytes.toBytes(Constants.RELATION_TABLE_CF2));
        Result result = relaTable.get(get);
        // 创建一个集合，存放微博内容表的put对象
        ArrayList<Put> inboxPuts = new ArrayList<>();
        // 遍历粉丝
        for (Cell cell : result.rawCells()) {
            // 构建微博收件箱表的put对象
            Put inboxPut = new Put(CellUtil.cloneQualifier(cell));
            // 给收件箱表的put对象赋值
            inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF), Bytes.toBytes(uid),
                    Bytes.toBytes(rowKey));
            // 将收件箱的put对象存入集合
            inboxPuts.add(inboxPut);
        }
        // 判断是否有粉丝
        if(inboxPuts.size() > 0){
            // 获取收件箱表对象
            Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
            // 插入收件箱表数据
            inboxTable.put(inboxPuts);
            inboxTable.close();
        }
        // 关闭资源
        relaTable.close();
        contTable.close();
        connection.close();
    }

    /**
     * 关注用户
     * @param uid
     * @param attentions
     * @throws IOException
     */
    public static void addAttentions(String uid, String... attentions) throws IOException {
        // 校验是否添加了待关注的人
        if(attentions.length <= 0){
            System.out.println("请选择待关注的人！");
            return;
        }
        Connection connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
        // 操作用户关系表
        Table relaTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
        // 用于存放关系表的put对象
        ArrayList<Put> relaPuts = new ArrayList<>();
        // 创建操作者的put对象
        Put uidPut = new Put(Bytes.toBytes(uid));
        // 循环创建被关注者的put对象
        for (String attention : attentions) {
            // 给操作者的put对象赋值
            uidPut.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF1), Bytes.toBytes(attention),
                    Bytes.toBytes(attention));
            // 创建被关注者的put对象
            Put attentPut = new Put(Bytes.toBytes(attention));
            // 给被关注者的put对象赋值
            attentPut.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF2),Bytes.toBytes(uid),
                    Bytes.toBytes(uid));
            // 将被关注者的put对象放入集合
            relaPuts.add(attentPut);
        }
        // 将操作者的put对象放入集合
        relaPuts.add(uidPut);
        // 用户关系表插入数据
        relaTable.put(relaPuts);
        // 操作收件箱表
        // 获取微博内容表对象
        Table contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
        // 创建收件箱表的put对象
        Put inboxPut = new Put(Bytes.toBytes(uid));
        // 循环attentions，获取每个被关注者的近期发布的微博
        for (String attention : attentions) {
            Scan scan = new Scan(Bytes.toBytes(attention + "_"), Bytes.toBytes(attention + "|"));
            ResultScanner resultScanner = contTable.getScanner(scan);
            Long ts = System.currentTimeMillis();
            for (Result result : resultScanner) {
                // 给收件箱表的put对象赋值
                inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF), Bytes.toBytes(attention),
                        ts++, result.getRow());
            }
        }
        // 判断当前的put对象是否为空
        if(!inboxPut.isEmpty()){
            Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
            // 插入数据
            inboxTable.put(inboxPut);
            inboxTable.close();
        }
        relaTable.close();
        contTable.close();
        connection.close();
    }

    /**
     * 取关用户
     * @param uid
     * @param dels
     * @throws IOException
     */
    public static void delAttentions(String uid, String... dels) throws IOException {
        if(dels.length <= 0){
            System.out.println("请添加待取关的用户！");
            return;
        }
        Connection connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
        // 操作用户关系表
        // 获取用户关系表对象
        Table relaTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
        // 创建集合存放用户关系表的delete对象
        ArrayList<Delete> relaDeletes = new ArrayList<>();
        // 创建操作者的delete对象
        Delete uidDelete = new Delete(Bytes.toBytes(uid));
        // 循环创建被取关者的delete对象
        for (String del : dels) {
            // 给操作者的delete对象赋值
            uidDelete.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF1), Bytes.toBytes(del));
            // 创建被取关者的delete对象
            Delete delete = new Delete(Bytes.toBytes(del));
            // 给被取关者的delete对象赋值
            delete.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF2), Bytes.toBytes(uid));
            // 将被取关者的delete对象添加到集合
            relaDeletes.add(delete);
        }
        // 将操作者的delete对象添加至集合
        relaDeletes.add(uidDelete);
        // 执行用户关系表的删除操作
        relaTable.delete(relaDeletes);
        // 操作收件箱表
        // 获取收件箱表对象
        Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
        // 创建操作者的delete对象
        Delete inboxDelete = new Delete(Bytes.toBytes(uid));
        // 给操作者的delete对象赋值
        for (String del : dels) {
            inboxDelete.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF), Bytes.toBytes(del));
        }
        // 执行收件箱表的删除操作
        inboxTable.delete(inboxDelete);
        relaTable.close();
        inboxTable.close();
        connection.close();
    }

    /**
     * 获取某个人的初始化页面数据
     * @param uid
     */
    public static void getInit(String uid) throws IOException {
        Connection connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
        // 获取收件箱表对象
        Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
        // 获取微博内容表对象
        Table contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
        // 创建收件箱表get对象，并获取数据（设置最大版本）
        Get inboxGet = new Get(Bytes.toBytes(uid));
        inboxGet.setMaxVersions();
        Result result = inboxTable.get(inboxGet);
        for (Cell cell : result.rawCells()) {
            // 构建微博内容表get对象
            Get contGet = new Get(CellUtil.cloneValue(cell));
            Result contResult = contTable.get(contGet);
            for (Cell contCell : contResult.rawCells()) {
                System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(contCell))
                        + ",CF:" + Bytes.toString(CellUtil.cloneFamily(contCell))
                        + ",CN:" + Bytes.toString(CellUtil.cloneQualifier(contCell))
                        + ",Value:" + Bytes.toString(CellUtil.cloneValue(contCell)));
            }
        }
        inboxTable.close();
        contTable.close();
        connection.close();
    }

    /**
     * 获取某个人的所有微博详情
     * @param uid
     */
    public static void getWeiBo(String uid) throws IOException {
        Connection connection = ConnectionFactory.createConnection(Constants.CONFIGURATION);
        // 获取收件箱表对象
        Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
        // 获取微博内容表对象
        Table contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
        // 构建scan对象
        Scan scan = new Scan();
        // 构建过滤器
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(uid + "_"));
        scan.setFilter(rowFilter);
        ResultScanner resultScanner = contTable.getScanner(scan);
        for (Result result : resultScanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(cell))
                        + ",CF:" + Bytes.toString(CellUtil.cloneFamily(cell))
                        + ",CN:" + Bytes.toString(CellUtil.cloneQualifier(cell))
                        + ",Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }
        contTable.close();
        connection.close();
    }



}
